package com.hadwinling.alogriithm.streaming

import org.apache.spark.SparkConf
import org.apache.spark.storage.StorageLevel
import org.apache.spark.streaming.dstream.ReceiverInputDStream
import org.apache.spark.streaming.receiver.Receiver
import org.apache.spark.streaming.{Seconds, StreamingContext}

import java.util.Random

/**
 * 自定义数据源需要继承 Receiver,并实现 onStart、onStop 方法来自定义数据源采集。
 */
object SparkStreaming03_DIY {
  def main(args: Array[String]): Unit = {
    //todo 创建环境
    val conf = new SparkConf().setMaster("local[*]").setAppName("SparkStreaming")
    val ssc = new StreamingContext(conf, Seconds(3))


    val messageDS: ReceiverInputDStream[String] = ssc.receiverStream(new MyReceiver())
    messageDS.print()

    ssc.start()
    ssc.awaitTermination()



  }
  /*
  自定义数据采集器
  1. 继承Receiver，定义泛型, 传递参数
  2. 重写方法
   */
  class MyReceiver extends Receiver[String](StorageLevel.MEMORY_ONLY) {
    private var flg = true
    //最初启动的时候,调用该方法,作用为:读数据并将数据发送给 Spark
    override def onStart(): Unit = {
      new Thread(new Runnable {
        override def run(): Unit = {
          while ( flg ) {
            val message = "采集的数据为：" + new Random().nextInt(10).toString
            store(message)
            Thread.sleep(500)
          }
        }
      }).start()
    }

    override def onStop(): Unit = {
      flg = false;
    }
  }

}
